package com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.utils;

import jakarta.annotation.Resource;
import org.apache.kafka.clients.admin.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.stream.Collectors;

@Component
public class KafkaAdminTools {

    @Resource
    private AdminClient adminClient;

    public CreateTopicsResult createTopic(String topicName, int partitions, short replicationFactor) throws Exception {
        Map<String, String> topicConfigs = new HashMap<>();
        Collection<TopicListing> existingTopics = adminClient.listTopics().listings().get();

        List<NewTopic> topics = new ArrayList<>();
        List<String> topicNames = existingTopics.stream().map(TopicListing::name).collect(Collectors.toList());
        if (!topicNames.contains(topicName)) {
            topics.add(new NewTopic(topicName, partitions, replicationFactor));
        }

        return adminClient.createTopics(topics);
    }
}
